package rpc_protobuf

import (
	"fmt"
	"log"
	"math/rand"
	"net"
	"net/rpc"
	"sync"
	"time"
)

// Go语言的RPC库最简单的使用方式是通过Client.Call方法进行同步阻塞调用
func doClientWork(client *rpc.Client) {
	helloCall := client.Go("HelloService.Hello", "hello", new(string), nil)

	// do smoe thing

	helloCall = <-helloCall.Done
	if err := helloCall.Error; err != nil {
		log.Fatal(err)
	}

	args := helloCall.Args.(string)
	reply := helloCall.Reply.(string)
	fmt.Println(args, reply)
}

// ------------------ rpc实现watch功能 --------
// 在很多系统中都提供了Watch监视功能的接口，当系统满足某种条件时Watch方法返回监控的结果。
// 在这里我们可以尝试通过RPC框架实现一个基本的Watch功能。
// 如前文所描述，因为client.send是线程安全的，我们也可以通过在不同的Goroutine中同时并发阻塞调用RPC方法。
// 通过在一个独立的Goroutine中调用Watch函数进行监控。

// 简单的内存KV数据库
// 其中m成员是一个map类型，用于存储KV数据。
// filter成员对应每个Watch调用时定义的过滤器函数列表
type KVStoreService struct {
	m      map[string]string
	filter map[string]func(key string)
	lock   sync.Mutex
}

type KVStoreServiceInterface interface {
	Get(key string, value *string) error
	Set(kv [2]string, reply *struct{}) error
	Watch(timeoutSecond int, keyChanged *string) error
}

func NewKVStoreService() *KVStoreService {
	return &KVStoreService{
		m:      map[string]string{},
		filter: map[string]func(key string){},
	}
}

func (s *KVStoreService) Get(key string, value *string) error {
	s.lock.Lock()
	defer s.lock.Unlock()

	if v, ok := s.m[key]; ok {
		*value = v
		return nil
	}

	return fmt.Errorf("not found")
}

// Set方法中，输入参数是key和value组成的数组，用一个匿名的空结构体表示忽略了输出参数
func (s *KVStoreService) Set(kv [2]string, reply *struct{}) error {
	s.lock.Lock()
	defer s.lock.Unlock()

	key, value := kv[0], kv[1]

	if oldValue := s.m[key]; oldValue != value {
		for _, fn := range s.filter {
			fn(key)
		}
	}

	s.m[key] = value
	return nil
}

func (s *KVStoreService) Watch(timeoutSecond int, keyChanged *string) error {
	id := fmt.Sprintf("watch-%s-%03d", time.Now(), rand.Int())
	ch := make(chan string, 10) // buffered

	s.lock.Lock()
	s.filter[id] = func(key string) {
		ch <- key
	}
	s.lock.Unlock()

	defer func() {
		// 删除监听
		s.lock.Lock()
		delete(s.filter, id)
		s.lock.Unlock()
	}()

	select {
	case <-time.After(time.Second * time.Duration(timeoutSecond)):
		*keyChanged = "超时"
		return nil
	case v := <-ch:
		*keyChanged = v
		return nil
	}
}

const KVStoreServiceName = "path/to/pkg.KVStoreService"

func RegisterKVStoreService(svc KVStoreServiceInterface) error {
	return rpc.RegisterName(KVStoreServiceName, svc)
}

func KVStoreRpcServer() {
	err := RegisterKVStoreService(NewKVStoreService())
	if err != nil {
		panic(err)
	}

	listen, err := net.Listen("tcp", ":1234")
	if err != nil {
		panic(err)
	}

	for {
		con, err := listen.Accept()
		if err != nil {
			panic(err)
		}

		go rpc.ServeConn(con)
	}
}

type KVStoreServiceClient struct {
	*rpc.Client
}

func (c *KVStoreServiceClient) Get(key string, value *string) error {
	return c.Call(KVStoreServiceName+".Get", key, value)
}

func (c *KVStoreServiceClient) Set(kv [2]string, reply *struct{}) error {
	return c.Call(KVStoreServiceName+".Set", kv, reply)
}

func (c *KVStoreServiceClient) Watch(timeoutSecond int, keyChanged *string) error {
	return c.Call(KVStoreServiceName+".Watch", timeoutSecond, keyChanged)
}

var _ KVStoreServiceInterface = (*KVStoreServiceClient)(nil)

func DialKVService(network, address string) (*KVStoreServiceClient, error) {
	client, err := rpc.Dial(network, address)
	if err != nil {
		panic(err)
	}

	return &KVStoreServiceClient{Client: client}, nil
}

func KVStoreRpcClient() {
	c, err := DialKVService("tcp", ":1234")
	if err != nil {
		panic(err)
	}

	go func() {
		var changed string
		err = c.Watch(3, &changed)

		fmt.Println("watch: " + changed)
	}()

	time.Sleep(time.Second * 1)

	err = c.Set([2]string{"abc", "changed-123"}, new(struct{}))
	if err != nil {
		panic(err)
	}

	time.Sleep(time.Second * 10)
}

//func doKVClientWork(c *KVStoreServiceClient) {
//	go func() {
//		var changed string
//		err := c.Watch(3, &changed)
//		if err != nil {
//			panic(err)
//		}
//
//		fmt.Println("watch: " + changed)
//	}()
//
//	time.Sleep(time.Second * 1)
//
//	err := c.Set([2]string{"abc", "changed-123"}, new(struct{}))
//	if err != nil {
//		panic(err)
//	}
//
//	time.Sleep(time.Second * 10)
//}

// ----------- 反向RPC -------------
// 通常的RPC是基于C/S结构，RPC的服务端对应网络的服务器，RPC的客户端也对应网络客户端
// 但是对于一些特殊场景，比如在公司内网提供一个RPC服务，但是在外网无法链接到内网的服务器。
// 这种时候我们可以参考类似反向代理的技术，首先从内网主动链接到外网的TCP服务器，然后基于TCP链接向外网提供RPC服务
func ReverseRpcServer() {
	rpc.Register(new(KVStoreServiceClient))

	for {
		con, err := net.Dial("tcp", ":1234")
		if err != nil {
			panic(err)
		}
		if con == nil {
			time.Sleep(time.Second * 1)
			continue
		}

		rpc.ServeConn(con)
		con.Close()
	}
}

func ReverseRpcClient() {
	listener, err := net.Listen("tcp", ":1234")
	if err != nil {
		panic(err)
	}

	clientChan := make(chan *rpc.Client)
	go func() {
		conn, err := listener.Accept()
		if err != nil {
			panic(err)
		}

		clientChan <- rpc.NewClient(conn)
	}()

	doReverseClientWork(clientChan)
}

// 首先从管道去取一个RPC客户端对象，并且通过defer语句指定在函数退出前关闭客户端。然后是执行正常的RPC调用
func doReverseClientWork(clientChan <-chan *rpc.Client) {
	client := <-clientChan
	defer client.Close()

	var reply string
	err := client.Call(KVStoreServiceName+".Get", "abc", reply)
	if err != nil {
		panic(err)
	}

	fmt.Println(reply)
}

// ---------------- 上下文信息 -----------------
// 基于上下文我们可以针对不同客户端提供定制化的RPC服务。我们可以通过为每个链接提供独立的RPC服务来实现对上下文特性的支持
func ContextRpcServer() {
	listener, err := net.Listen("tcp", ":1234")
	if err != nil {
		log.Fatal("ListenTCP error:", err)
	}

	for {
		conn, err := listener.Accept()
		if err != nil {
			panic(err)
		}

		go func() {
			defer conn.Close()

			r := rpc.NewServer()
			r.Register(&HelloService{conn: conn})
			r.ServeConn(conn)
		}()
	}
}
